Max/asyncread asyncwrite nym client #6318
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
| ) -> Result<Vec<u64>> { | ||
| let mut request_ids = Vec::with_capacity(3); | ||
|
|
||
| for _ in 1..=3 { |
There was a problem hiding this comment.
note that this changes semantics of the original call. the former was performing all 3 pings concurrently. the updated variant does it sequentially - make sure this is what you want
| self | ||
| } | ||
| #[allow(clippy::expect_used)] | ||
| pub fn serialized_size(&self) -> u64 { |
There was a problem hiding this comment.
this doesn't seem to be used anywhere
| } | ||
| } | ||
|
|
||
| pub struct InputMessageCodec; |
There was a problem hiding this comment.
are we sure InputMessage is the right abstraction for the AsyncRead/AsyncWrite impl? if you're concerned about reading/writing bytes, why should you care about internal TransmissionLane (that's only used for backpressure)?
wouldn't it be sufficient to just use the inbuilt LengthDelimitedCodec wrapped around simple Vec<u8> and then call normal "send" within the client on that message?
| } | ||
|
|
||
| // MAX TODO implement for v1 as well for back compat? - this was added in the original asyncread/write work when we only had one v | ||
| impl Serialize for MixPacket { |
There was a problem hiding this comment.
given we have to implement serde for internal MixPacket makes me think we chose wrong abstraction. client really shouldn't be concerned about this guy. it's just a wrapper for data that's eventually going to get encoded in websocket messages for the gateway
| // TEMP UNTIL FURTHER REFACTORING | ||
| pub use preparer::payload::NymPayloadBuilder; | ||
|
|
||
| fn make_bincode_serializer() -> impl bincode::Options { |
There was a problem hiding this comment.
not sure how to feel about using bincode inside sphinx itself...
| Poll::Ready(Ok(())) | ||
| } else { | ||
| let written = buf.capacity(); | ||
| buf.put_slice(&self._read.buffer[..written]); |
There was a problem hiding this comment.
use split method on the BytesMut
|
|
||
| let msg_size = msg.serialized_size(); | ||
|
|
||
| let mut fut = pin!(self.client_input.send(msg)); |
There was a problem hiding this comment.
you're creating a new send future every single time you call poll_write without letting the previous one resolve. you will have to actually store the future on the MixnetClient itself and check for its existence in subsequent calls
There was a problem hiding this comment.
just a random thought (don't know if it's a good idea!): what if AsyncRead / AsyncWrite was changed to the simplest possible variant where they only read/write bytes to internal client buffers without any encoding. but there would be a dedicated internal client task/thread that periodically would check those buffers and send/recv from the mixnet on a timer based on their contents
| self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| ) -> Poll<std::prelude::v1::Result<(), std::io::Error>> { | ||
| Sink::poll_flush(self, cx).map_err(|_| std::io::Error::other("failed to flush the sink")) |
There was a problem hiding this comment.
I'm not sure if calling the internal Sink impl for this guy is a good call
| self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| ) -> Poll<std::prelude::v1::Result<(), std::io::Error>> { | ||
| AsyncWrite::poll_flush(self, cx) |
There was a problem hiding this comment.
if we don't have anything substantial to do here, it's fine to just return an Ok(()) straightaway
|
|
||
| let msg_size = msg.serialized_size(); | ||
|
|
||
| let mut fut = pin!(self.client_input.send(msg)); |
There was a problem hiding this comment.
same problem as with AsyncRead - this future will never have time to resolve unless it was already ready and won't wake the waker
mfahampshire
modified the milestones:
Oscypek (est. mainnet 27.01.2026),
Parmigiano - Mainnet release 10th Feb
Breaking up #6129 into more reasonably sized chunks.
This is the first which is basically the original addition of
AsyncRead/AsyncWriteto the core client.This change is